package com.shujia.spark.streaming

import java.util.Properties

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo5Bro {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[2]").setAppName("wc")
    conf.set("spark.sql.shuffle.partitions", "1")

    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)

    //创建sparkStraming上下文对象,  需要指定创建RDD的间隔时间    batch时间   可以理解为多少秒计算一次
    val ssc = new StreamingContext(sc, Durations.seconds(5))

    val ds = ssc.socketTextStream("node1", 8888)

    //黑名单
    //    val black = List("D55433A437AEC8D8D3DB2BCA56E9E64392A9D93C", "47BE1E866CFC071DB19D5E1C056BE28AE24C16E7")

    /**
      * create table black(mdn varchar(100));
      *
      */
    /*

        //连接mysql获取黑名单
        val properties = new Properties()

        properties.setProperty("driver", "com.mysql.jdbc.Driver")
        properties.setProperty("user", "test")
        properties.setProperty("password", "123456")
        val blackDF = sqlContext.read.jdbc("jdbc:mysql://node1:3306/test", "black", properties)
        //DF --->  RDD
        val blackRDD = blackDF.map(row => {
          //取出手机号
          val mdn = row.getAs[String]("mdn")
          mdn
        })

        //将数据拉取到Drveri端
        val black = blackRDD.collect().toList

        println("黑名单：" + black)

        //广播黑名单
        val blackBro = sc.broadcast(black)

        //过滤黑名单数据
        val blackDS = ds.filter(line => {

          //手机号
          val mdn = line.split(",")(0)

          //判断是否在黑名单中
          blackBro.value.contains(mdn)
        })
    */

    /**
      * 动态修改广播变量
      *
      */

    val blackDS = ds.foreachRDD(rdd => {
      println("driver端代码，每一个batch运行一次")
      //连接mysql获取黑名单
      val properties = new Properties()

      properties.setProperty("driver", "com.mysql.jdbc.Driver")
      properties.setProperty("user", "test")
      properties.setProperty("password", "123456")
      val blackDF = sqlContext.read.jdbc("jdbc:mysql://node1:3306/test", "black", properties)
      //DF --->  RDD
      val blackRDD = blackDF.map(row => {
        //取出手机号
        val mdn = row.getAs[String]("mdn")
        mdn
      })

      //将数据拉取到Drveri端
      val black = blackRDD.collect().toList

      println("黑名单：" + black)

      //广播黑名单,  每一个batch都会广播一次
      val blackBro = sc.broadcast(black)


      val rdd1 = rdd.filter(line => {

        //手机号
        val mdn = line.split(",")(0)

        //判断是否在黑名单中
        blackBro.value.contains(mdn)
      })

      //报警
      rdd1.foreach(println)

      //清除广播变量的数据 ，  清空广播变量的操作需要放在action算子后面
      blackBro.destroy()
    })


    //启动sparkstreaming程序
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()


  }
}
